package com.gitee.xhs;

import com.gitee.xhs.customFunction.*;
import com.gitee.xhs.dao.IterateTransaction;
import com.gitee.xhs.dao.IntervalJoinDao;
import com.gitee.xhs.joining.intervalJoin.IntervalJoinMain;
import com.gitee.xhs.iterateStream.IterateStreamMain;
import com.gitee.xhs.sampleRecod.transactionRecord.Transaction;
import com.gitee.xhs.window.evictors.SimpleEvictors;
import com.gitee.xhs.window.lateness.SimpleLatenessMain;
import com.gitee.xhs.window.triggers.SimpleWindowTriggerMain;
import com.gitee.xhs.window.windowOperator.SimpleWindowMain;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author BigHead
 * @date 2022/7/19 10:37
 * @description OperatorMain
 */
public class OperatorMain {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment streamExecutionEnvironment = FlinkEnvironmentFactory.webUi(new EnvironmentConfig(null));

        // 主流
        SingleOutputStreamOperator<Transaction> mainDataStream = streamExecutionEnvironment
                .addSource(
                        OperatorSourceFactory.unBoundLoopOneToTenOneSecondSource()
                )
                .map(
                        new OperatorMap()
                )
                .flatMap(
                        new OperatorFlatMap()
                )
                .filter(
                        new OperatorFilter()
                )
                .keyBy(
                        new OperatorKeyBy()
                )
                .reduce(
                        new OperatorReduce()
                )
                .process(
                        new OperatorCopyTransactionProcess()
                );

//        mainDataStream
//                .print("主流效果");

        // Iterate 效果
        IterateStreamMain iterateStreamMain = new IterateStreamMain();

        SingleOutputStreamOperator<IterateTransaction> iterateTransactionStream = iterateStreamMain
                .iterateLogic(mainDataStream);

//        iterateTransactionStream
//                .print("Iterate 效果");

        // connect 效果
        DataStream<Transaction> connectStream = mainDataStream
                .connect(iterateTransactionStream)
                .map(
                        new OperatorCoMap()
                )
                .keyBy(
                        new OperatorKeyBy()
                ).process(
                        new OperatorSeparationTransactionProcess()
                );

//        connectStream
//                .print("connect 效果");

        // Union 效果
        DataStream<Transaction> unionSource = streamExecutionEnvironment
                .addSource(
                        OperatorSourceFactory.unBoundLoopTenToTwentyOneSecondSource()
                )
                .map(
                        new OperatorUnionMap()
                );

        DataStream<Transaction> simpleEndStream = connectStream
                .union(unionSource);

//        simpleEndStream
//                .print("Union 效果");

        // intervalJoin 效果
        IntervalJoinMain intervalJoinMain = new IntervalJoinMain();

        SingleOutputStreamOperator<IntervalJoinDao> intervalJoinDaoStream = intervalJoinMain
                .intervalJoinLogic(
                        streamExecutionEnvironment,
                        simpleEndStream
                );

//        intervalJoinDaoStream
//                .print("intervalJoin 效果");

        // window
        SimpleWindowMain simpleWindowMain = new SimpleWindowMain(streamExecutionEnvironment);
//        simpleWindowMain.windowLogic();

        // window trigger
        SimpleWindowTriggerMain simpleWindowTriggerMain = new SimpleWindowTriggerMain(streamExecutionEnvironment);
//        simpleWindowTriggerMain.triggerLogic();

        // window evictors
        SimpleEvictors simpleEvictors = new SimpleEvictors(streamExecutionEnvironment);
//        simpleEvictors.evictorsLogic();

        // window Lateness
        SimpleLatenessMain simpleLatenessMain = new SimpleLatenessMain(streamExecutionEnvironment);
        simpleLatenessMain.latenessLogic();

        streamExecutionEnvironment.execute("OperatorMain");
    }

}
